package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo05GroupBy {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext

    val conf: SparkConf = new SparkConf()

    conf.setAppName("Demo05GroupBy")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val lineRDD: RDD[String] = sc.parallelize(List("hadoop,hive,hbase", "mapreduce,spark,flink", "mapreduce,spark,flink"))

    // 统计每个单词的数量
    val wordsRDD: RDD[String] = lineRDD
      .flatMap(line => line.split(","))

    // 分组
    /**
     * groupBy 转换算子
     * 同List中的groupBy方法类似
     * 需要指定一个分组字段
     * 最终会返回一个RDD 不像List中的groupBy会返回一个Map
     * 但RDD中每条数据的类型也是一个二元组（可以看成kv格式）
     * “Key” 指定的分组字段
     * “value” 属于同一个分组的所有的数据
     */
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD
      .groupBy(word => word)


    groupRDD
      .map(kv => s"${kv._1},${kv._2.size}")
      .foreach(println)


  }

}
